package cn.com.zetatech.loader.common.service.impl;

import cn.com.zetatech.loader.api.exception.LoaderException;
import cn.com.zetatech.loader.common.service.KafkaService;
import cn.com.zetatech.loader.common.service.ResolverStateControlService;
import cn.com.zetatech.loader.common.service.ResolverStoreService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * @author zhengbs
 */
@Service
@Slf4j
public class ResolverStoreServiceImpl implements ResolverStoreService {

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private ResolverStateControlService resolverStateControlService;

    /**
     * 保存数据时调用
     * 1. 发送数据到kafka
     * 2. 添加数据状态监控
     * @param topic  发送的kafka topic
     * @param data   待保存的数据json
     * @param fileId 解析文件ID
     * @param resolverCompleted 文件是否解析完
     * @return
     */
    @Override
    public boolean dataStore(String topic, String data, String fileId, boolean resolverCompleted, String complateInfo) {
        if(resolverStateControlService.checkHasError(fileId)) {
            log.info("已发生错误，忽略发送topic[{}]数据，fileId：{}", topic, fileId);
            throw new LoaderException("已发生错误，忽略发送topic[{}]数据，fileId：" + fileId);
        } else {
            RecordMetadata kafkaRecordMetadata = sendMessage(topic, data, fileId);
            log.info("发送topic[{}]数据，fileId：{}，partition：{}，offset：{}", topic, fileId, kafkaRecordMetadata.partition(), kafkaRecordMetadata.offset());
            resolverStateControlService.dataStoreHandle(topic, kafkaRecordMetadata.partition(), kafkaRecordMetadata.offset(), fileId);
        }
        if(resolverCompleted) {
            resolverStateControlService.completedHandle(fileId, complateInfo);
        }
        return true;
    }


    private RecordMetadata sendMessage(String topic, String message, String fileId) {
        try {
            List<Header> headers = Collections.singletonList(new RecordHeader("loaderFileId", fileId.getBytes()));
            return kafkaService.sendMessage(topic, message, headers);
        } catch (Exception e) {
            log.error(e.getMessage(),e);
            throw new RuntimeException(e);
        }
    }

}
